package hit.edu.zjc.Operator;


import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.IOperatorDescriptorRegistry;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
/*
 * This Class AFK
 * */
public class SourceByteWriterOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
    private static final long serialVersionUID = 1L;

    private IFileSplitProvider fileSplitProvider;

    private byte[] data;
    
    public SourceByteWriterOperatorDescriptor(IOperatorDescriptorRegistry spec, IFileSplitProvider fileSplitProvider,byte[] bytedata) {
        super(spec, 0, 0);
        this.fileSplitProvider = fileSplitProvider;
        data=bytedata;
    }


    @Override
    public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, final int partition, int nPartitions) {
    	final FileSplit[] splits = fileSplitProvider.getFileSplits();
    	return new AbstractUnaryOutputSourceOperatorNodePushable() {
    		
    		  private OutputStream out;
            @Override
            public void initialize() throws HyracksDataException {
             try 
                {
                 out = new FileOutputStream(splits[partition].getLocalFile().getFile());
                 out.write(data);
                } 
             catch (IOException e) 
                {
                 throw new HyracksDataException(e);
                }
             
            	
            }
        };
    }
}